package com.ruoyi.yanjing.client;

import com.ruoyi.common.core.domain.AjaxResult;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import java.util.concurrent.ConcurrentSkipListSet;

import static com.ruoyi.common.core.domain.AjaxResult.error;
import static com.ruoyi.common.core.domain.AjaxResult.success;

/**
 * MQTT Push Client
 */
@Component
public class MqttPushClient {
    private static final Logger logger = LoggerFactory.getLogger(MqttPushClient.class);
    @Value("${spring.mqtt.hostUrl}")
    private String hostUrl;

    @Value("${spring.mqtt.clientId}")
    private String clientId;

    @Value("${spring.mqtt.username}")
    private String username;

    @Value("${spring.mqtt.password}")
    private String password;

    @Value("${spring.mqtt.timeout}")
    private int timeout;

    @Value("${spring.mqtt.keepalive}")
    private int keepalive;
    @Autowired
    private PushCallback pushCallback;

    @Autowired
    private static MqttClient client;

    // 用于记录已订阅的主题
    private static final ConcurrentSkipListSet<String> subscribedTopics = new ConcurrentSkipListSet<>();

    private static MqttClient getClient() {
        return client;
    }

    private static void setClient(MqttClient client) {
        MqttPushClient.client = client;
    }

    /**
     * Connect to MQTT broker.
     *
     * @param host      MQTT broker host address.
     * @param clientID  Client ID.
     * @param username  Username for authentication.
     * @param password  Password for authentication.
     * @param timeout   Connection timeout.
     * @param keepalive Keep alive interval.
     */
//    public void connect(String host, String clientID, String username, String password, int timeout, int keepalive) {
//        try {
//            client = new MqttClient(host, clientID, new MemoryPersistence());
//            MqttConnectOptions options = createConnectOptions(username, password, timeout, keepalive);
//            setClient(client);
//            client.setCallback(new PushCallback());
//            client.connect(options);
//            logger.info("Connected to MQTT broker: {}", host);
//        } catch (MqttException e) {
//            logger.error("Failed to connect to MQTT broker", e);
//        }
//    }
    public void connect(String host, String clientID, String username, String password, int timeout, int keepalive) {
        try {
            client = new MqttClient(host, clientID, new MemoryPersistence());
            MqttConnectOptions options = createConnectOptions(username, password, timeout, keepalive);
            setClient(client);
            client.setCallback(pushCallback);
            client.connect(options);
            logger.info("Connected to MQTT broker: {}", host);
        } catch (MqttException e) {
            logger.error("Failed to connect to MQTT broker", e);
            throw new RuntimeException(e);
        }
    }

    public void connectToBroker() {
        connect(hostUrl, clientId, username, password, timeout, keepalive);
        resubscribeTopics();
    }

    private void resubscribeTopics() {
        for (String topic : subscribedTopics) {
            subscribe(topic, 0);
        }
    }

    private MqttConnectOptions createConnectOptions(String username, String password, int timeout, int keepalive) {
        MqttConnectOptions options = new MqttConnectOptions();
        options.setCleanSession(true);
        options.setUserName(username);
        options.setPassword(password.toCharArray());
        options.setConnectionTimeout(timeout);
        options.setKeepAliveInterval(keepalive);
        return options;
    }


    /**
     * Publish a message to a topic.
     *
     * @param qos         Quality of Service level.
     * @param retained    Whether the message should be retained.
     * @param topic       Topic to publish to.
     * @param pushMessage Message payload.
     * @return AjaxResult indicating success or failure.
     */
    public AjaxResult publish(int qos, boolean retained, String topic, String pushMessage) {
        MqttMessage message = createMqttMessage(qos, retained, pushMessage);
        MqttTopic mqttTopic = getClient().getTopic(topic);
        if (mqttTopic == null) {
            logger.error("Topic does not exist: {}", topic);
            return error("Topic does not exist");
        }
        try {
            MqttDeliveryToken token = mqttTopic.publish(message);
            token.waitForCompletion();
            logger.info("Message published to topic: {}, Payload: {}", topic, pushMessage);
            return success("Message published successfully");
        } catch (MqttException e) {
            logger.error("Failed to publish message", e);
            return error("Failed to publish message");
        }
    }

    private MqttMessage createMqttMessage(int qos, boolean retained, String pushMessage) {
        MqttMessage message = new MqttMessage(pushMessage.getBytes());
        message.setQos(qos);
        message.setRetained(retained);
        return message;
    }

    /**
     * Subscribe to a topic dynamically.
     *
     * @param topic Topic to subscribe to.
     * @param qos   Quality of Service level.
     */
    public void subscribe(String topic, int qos) {
        if (subscribedTopics.contains(topic)) {
            logger.info("Already subscribed to topic: {}", topic);
            return;
        }
        try {
            getClient().subscribe(topic, qos);
            subscribedTopics.add(topic);
            logger.info("Successfully subscribed to topic: {}", topic);
        } catch (MqttException e) {
            logger.error("Failed to subscribe to topic: {}", topic, e);
        }
    }
}
